package com.example.es.controller;

import com.example.es.pojo.Products;
import com.example.es.service.ProductsService;
import com.google.gson.Gson;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.xcontent.XContentType;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.io.IOException;

@Component
public class ProductConsumer {
    @Autowired
    private ProductsService productsService;
    @Autowired
    private RestHighLevelClient restHighLevelClient;
    private Gson gson = new Gson();
    @KafkaListener(topics = "product")
    public void listen(ConsumerRecord<?, ?> record) throws IOException {
        Products products = productsService.findById(Integer.parseInt(record.value() + ""));
        String source =gson.toJson(products);
        IndexRequest request = new IndexRequest("lagou","product",products.getId()+"");
        request.source(source, XContentType.JSON);
        IndexResponse response = restHighLevelClient.index(request, RequestOptions.DEFAULT);
        System.out.println(response);
    }
}
